package skewreduce.pig;

import java.io.EOFException;
import java.io.IOException;
import java.net.URL;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.ByteWritable;
import org.apache.hadoop.io.SequenceFile;

import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.logicalLayer.schema.Schema;

import eagg.framework.MuxData;

public class MuxStorage extends EaggLoadBase {
    private byte targetStream;
    private SequenceFile.Reader reader;
    private long end;

    private ByteWritable streamID = new ByteWritable();
    private MuxData value = new MuxData();

    public MuxStorage() {}

    public MuxStorage(int stream,String format) {
        super(format);
        targetStream = (byte)stream;
    }

    public MuxStorage(String stream,String format) {
        super(format);
        targetStream = Byte.parseByte(stream);
    }

    public void bindTo(String fileName,
                       BufferedPositionedInputStream is,
                       long offset,
                       long end) throws IOException {
        inferReader(fileName);
        if ( offset != 0 )
            reader.sync(offset);

        this.end = end;
    }

    public Tuple getNext() throws IOException {
        while ( true ) {
            if ( reader != null && (reader.getPosition() < end || ! reader.syncSeen()) && reader.next(streamID,value) ) {
                if ( targetStream != streamID.get() ) continue;
                // we are good to parse...
                try {
                    return createTuple(value.getDataInput());
                } catch ( EOFException eof ) {
                    System.err.println("Partially written record?!");
                    return null;
                }
            }
            return null;
        }
    }

    protected void inferReader(String fileName) throws IOException {
        if ( reader == null ) {
            Configuration conf = new Configuration();
            final Path path = new Path(fileName);
            FileSystem fs = FileSystem.get(path.toUri(),conf);
            reader = new SequenceFile.Reader(fs,path,conf);
        }
    }
}
